package com.qupeng.demo.kafka.kafkaapache.security;

import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
import org.apache.kafka.common.utils.Utils;

import javax.security.auth.callback.Callback;
import javax.security.auth.callback.NameCallback;
import javax.security.auth.callback.PasswordCallback;
import javax.security.auth.callback.UnsupportedCallbackException;
import javax.security.auth.login.AppConfigurationEntry;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
import static org.apache.kafka.common.config.ConfigDef.Type.PASSWORD;
import static org.apache.kafka.common.config.ConfigDef.Type.STRING;

public class PasswordProvider implements AuthenticateCallbackHandler {

    @Override
    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
        Properties props = Utils.loadProps("src/main/resources/password.properties");
        PasswordConfig config = new PasswordConfig(props);
        String user = config.getString("username");
        String password = config.getPassword("password").value();
        for (Callback callback : callbacks) {
            if (callback instanceof NameCallback)
                ((NameCallback) callback).setName(user);
            else if (callback instanceof PasswordCallback) {
                ((PasswordCallback) callback).setPassword(password.toCharArray());
            }
        }
    }

    @Override
    public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {

    }

    @Override
    public void close() {

    }

    private static class PasswordConfig extends AbstractConfig {
        static ConfigDef CONFIG = new ConfigDef()
                .define("username", STRING, HIGH, "User name")
                .define("password", PASSWORD, HIGH, "User password");

        PasswordConfig(Properties props) {
            super(CONFIG, props, false);

        }
    }
}
